Membrane Guide

v0.9.0

Membrane Framework guide

Hello, and welcome to the general guide to the Membrane Framework!

Membrane is a multimedia processing framework that focuses on reliability, concurrency, and scalability. It is primarily written in Elixir, while some platform-specific or time-constrained parts are written in C. With a range of existing packages and an easy-to-use interface for writing your own, Membrane can be used to process almost any type of multimedia.

Structure of the Membrane Framework

The Membrane Framework allows you to define your own processing units, called "elements." An element might be able to mux incoming audio and video streams into a container, or play raw audio using your sound card.

Elements can be organized into a "pipeline," - a sequence of linked elements that performs a specific task. For example, a pipeline might receive an incoming RTSP stream from a webcam and convert it to an HLS stream, or act as a selective forwarding unit (SFU) to implement your own videoconferencing room.

Membrane Core

Membrane Core is the heart and soul of the Membrane Framework. It is written entirely in Elixir and provides the internal mechanisms and developer's interfaces that allow you to prepare processing elements and link them together in a convenient yet reliable way. Note that Membrane Core does not contain any multimedia-specific logic. The documentation for the developer's API is available at hexdocs.

Membrane packages

Membrane packages provide the multimedia processing logic. They are organized by domain, so a package named after a container (e.g. membrane_mp4_plugin) contains elements for muxing and demuxing that container. A complete list of all the Membrane packages managed by the Membrane team is available here.

A complete list of all the Membrane packages managed by the Membrane team is available here.

Where can I learn Membrane?

There are a number of resources available for learning about Membrane:

This guide

The following sections in that guide will introduce the main concepts of creating Membrane elements and pipelines, without focusing on the specific details of multimedia processing.

Demos

The membrane demos are available in a repository with multiple subdirectories, each containing a single Elixir project. Each subdirectory includes instructions for running the demo application, but you will need to examine the code to understand how it works. To access the demos, follow this link.

Tutorials

For a step-by-step guide to implementing a specific system using Membrane, check out our tutorials.

Documentation

For more detailed information, you can refer to the Membrane Core documentation and the documentation for the Membrane packages maintained by the Membrane team, both of which can be accessed from here.

If you see something requiring improvement in this guide, feel free to create an issue or open a PR in this repository. For more information about the framework and more tutorials, check our website: membrane.stream.

Demos

The official Membrane demos reside in the membrane demo repository.

Basics

Demo name | Description | Links
simple_element | An example of a simple Membrane's element capable of counting the incoming buffers. | GitHub
simple_pipeline | An example of a Membrane's pipeline playing an mp3 file. | GitHub

HLS

Demo name | Description | Links
camera_to_hls | A demonstration of capturing camera output and converting it to an HLS stream. | GitHub
rtmp_to_hls | Receiving RTMP stream and broadcasting it via HLS. | GitHub
rtp_to_hls | Receiving RTP stream and broadcasting it via HLS. | GitHub
rtsp_to_hls | Receiving RTSP stream and converting it to HLS. | GitHub
webrtc_to_hls | Converting WebRTC stream into HLS. | GitHub

WebRTC

Demo name | Description | Links
webrtc_videoroom | A basic example of Membrane RTC Engine. It's as simple as possible just to show you how to use our API. | GitHub
webrtc_simple | A simple example of signalling server based on Membrane.WebRTC.Server. | GitHub
webrtc_athentication | An example of signalling server with authentication based on Membrane.WebRTC.Server. | GitHub

Other

Demo name | Description | Links
rtp | Sending and receiving RTP/SRTP stream. | GitHub
video_mixer | How to mix audio and video files. | GitHub

Membrane packages

If you see any out-of-date information or an invalid link here, feel free to create an issue or open a PR in the guide repository

To improve modularity, Membrane is split into multiple packages. Here we gathered all of the official ones and keep adding third-party ones too, so if you created a Membrane plugin or other Membrane-based library, let us know! We use the following [tags] in package descriptions:

[In progress] - in active development, but not yet ready

[Experimental] - possible to use, but may lack some core features/tests/documentation

[Alpha] - ready to use, but may lack some features or need more testing

[Third-party] - maintained by the community

[Suspended] - not ready and development was suspended

Check the Membrane GitHub organization to see which packages are actively developed currently.

Demos

The official Membrane demos reside in the membrane demo repository. Check there for details.

Docker

The Membrane docker (Docker Hub, GitHub) contains Erlang, Elixir and libraries necessary to test and run the Membrane Framework.

Media agnostic packages

Package | Description | Links
membrane_core | The core of the framework | Docs GitHub
membrane_common_c | Utilities for the native parts of Membrane | Docs GitHub
membrane_telemetry_metrics | Tool for generating metrics | Docs GitHub
membrane_opentelemetry | Wrapper of OpenTelemetry functions for Membrane Multimedia Framework | Docs GitHub
bundlex | Tool for compiling C/C++ code within Mix projects | Docs GitHub
unifex | Tool automatically generating NIF and CNode interfaces between C/C++ and Elixir | Docs GitHub
bunch | Extension of Elixir standard library | Docs GitHub
sebex | The ultimate assistant in Membrane Framework releasing & development | GitHub
Beamchmark | Tool for measuring BEAM performance | Docs GitHub

Plugins

Membrane plugins are packages which provide the elements and bins that can be used in pipelines built with the Membrane Framework. These plugins can provide various multimedia processing capabilities and help in dealing with various formats, codecs, protocols, containers and external APIs.

Media agnostic

Package | Description | Links
membrane_file_plugin | Plugin for reading and writing to files | Docs GitHub
membrane_udp_plugin | Plugin for sending and receiving UDP streams | Docs
membrane_hackney_plugin | HTTP sink and source based on Hackney library | Docs GitHub
membrane_scissors_plugin | Element for cutting off parts of the stream | Docs GitHub
membrane_tee_plugin | Plugin for splitting data from a single input to multiple outputs | Docs GitHub
membrane_funnel_plugin | Plugin for merging multiple input streams into a single output | Docs GitHub
membrane_realtimer_plugin | Membrane element limiting playback speed to realtime, according to buffers' timestamps | Docs GitHub
membrane_stream_plugin | Plugin for dumping and restoring a Membrane Stream to and from a binary format | Docs GitHub
membrane_fake_plugin | Plugin with fake sink elements that consume & drop incoming data | Docs GitHub
membrane_pcap_plugin | [Experimental] Plugin for reading files in pcap format | GitHub
membrane_live_framerate_converter_plugin | [Third-Party] [Experimental] Plugin for producing a stable output framerate | GitHub
membrane_node_proxy | [Third-Party] [Experimental] Plugin providing a mechanism for starting and connecting sources and sinks that span Erlang nodes | GitHub

Media network protocols & containers

Package | Description | Links
membrane_rtp_plugin | Membrane bins and elements for handling RTP and RTCP streams | Docs GitHub
membrane_rtp_h264_plugin | RTP payloader and depayloader for H264 | Docs GitHub
membrane_rtp_vp8_plugin | RTP payloader and depayloader for VP8 | Docs GitHub
membrane_rtp_vp9_plugin | RTP payloader and depayloader for VP9 | GitHub
membrane_rtp_mpegaudio_plugin | RTP MPEG Audio depayloader | Docs GitHub
membrane_rtp_opus_plugin | RTP payloader and depayloader for OPUS audio | Docs GitHub
membrane_rtmp_plugin | Plugin for receiving, demuxing, parsing and streaming RTMP streams. Includes TCP server for handling connections | Docs GitHub
membrane_mpegts_plugin | MPEG-TS demuxer | Docs GitHub
membrane_mp4_plugin | Utilities for MP4 container parsing and serialization and elements for muxing the stream to CMAF | Docs GitHub
membrane_http_adaptive_stream_plugin | Plugin generating manifests for HLS (DASH support planned) | Docs GitHub
membrane_matroska_plugin | Plugin for muxing audio and video streams to Matroska container and demuxing matroska stream to audio and video | Docs GitHub
membrane_flv_plugin | Plugin for muxing audio and video streams to FLV format and demuxing FLV stream to audio and video | Docs GitHub
membrane_ivf_plugin | Plugin for serializing and deserializng Indeo Video Format stream | Docs GitHub
membrane_ogg_plugin | [Experimental] Ogg demuxer and parser | GitHub
membrane_quic_plugin | [Experimental] Plugin containing elements for sending and receiving data over QUIC. It contains also QUIC server | GitHub
membrane_hls_plugin | [Third-Party] [Alpha] Membrane.HLS.Source element for HTTP Live Streaming (HLS) playlist files | GitHub
membrane_ogg_plugin | [Third-Party] [Alpha] Ogg plugin based on the libogg | GitHub

Audio codecs

Package | Description | Links
membrane_aac_plugin | AAC parser and complementary elements for AAC codec | Docs GitHub
membrane_aac_fdk_plugin | AAC decoder and encoder based on FDK library | Docs GitHub
membrane_flac_plugin | Parser for files in FLAC bitstream format | Docs GitHub
membrane_mp3_lame_plugin | Membrane MP3 encoder based on Lame | Docs GitHub
membrane_mp3_mad_plugin | Membrane MP3 decoder based on MAD | Docs GitHub
membrane_element_mpegaudioparse | Element capable of parsing bytestream into MPEG audio frames | Docs GitHub
membrane_opus_plugin | Opus encoder, decoder and parser | Docs GitHub
membrane_wav_plugin | WAV parser | Docs GitHub

Video codecs

Package | Description | Links
membrane_h264_ffmpeg_plugin | Membrane H264 parser, decoder and encoder based on FFmpeg and x264 | Docs GitHub
membrane_h264_plugin | Pure Elixir Membrane H264 parser | Docs GitHub
turbojpeg | [Third-party] libjpeg-turbo bindings for Elixir by Binary Noggin | Docs GitHub
membrane_subtitle_mixer_plugin | [Third-party] [Alpha] Plugin that uses CEA708 to merge subtitles directly in H264 packets | GitHub
membrane_mpeg_ts_plugin | [Third-party] [Alpha] MPEG-TS Demuxer | GitHub

Raw

Package | Description | Links
membrane_generator_plugin | Video and audio samples generator | Docs GitHub

Raw audio

Package | Description | Links
membrane_portaudio_plugin | Raw audio retriever and player based on PortAudio | Docs GitHub
membrane_ffmpeg_swresample_plugin | Plugin performing audio conversion, resampling and channel mixing, using SWResample module of FFmpeg library | Docs GitHub
membrane_audiometer_plugin | Elements for measuring the level of the audio stream | Docs GitHub
membrane_audio_mix_plugin | Elements for mixing audio streams | Docs GitHub
membrane_element_fade | [Experimental] | GitHub

Raw video

Package | Description | Links
membrane_sdl_plugin | Membrane video player based on SDL | Docs GitHub
membrane_raw_video_parser_plugin | Plugin for parsing raw video streams | Docs GitHub
membrane_ffmpeg_swscale_plugin | Plugin for scaling raw video streams and converting raw video format | Docs GitHub
membrane_framerate_converter_plugin | Plugin for adjusting the framerate of raw video stream | Docs GitHub
membrane_video_merger_plugin | Plugin for cutting and merging raw video streams | Docs GitHub
membrane_camera_capture_plugin | This plugin can be used to capture video stream from an input device | Docs GitHub
membrane_ffmpeg_video_filter_plugin | This plugin contains elements providing video filters based on ffmpeg video filter feature | Docs GitHub
membrane_video_compositor_plugin | [Alpha] WGPU based plugin for composing, transforming and editing multiple raw videos in real time | GitHub
membrane_video_mixer_plugin | [Experimental] [Third-Party] Element which mixes multiple video inputs to a single output using ffmpeg filters | GitHub

External APIs

Package | Description | Links
membrane_element_gcloud_speech_to_text | Plugin providing speech recognition via Google Cloud Speech-to-Text API | Docs GitHub
membrane_element_ibm_speech_to_text | Plugin providing speech recognition via IBM Cloud Speech-to-Text service | Docs GitHub
membrane_s3_plugin | [Third-Party] Plugin providing sink that writes to Amazon S3 | Docs GitHub
membrane_transcription | [Third-Party] [Experimental] Plugin based on Whisper allowing creating transcriptions | GitHub
xturn_membrane | [Third-Party] [Experimental] Experimental Membrane source / sink for XTurn | GitHub

Formats

Package | Description | Links
membrane_raw_audio_format | Raw audio format definition | Docs GitHub
membrane_raw_video_format | Raw video format definition | Docs GitHub
membrane_aac_format | Advanced Audio Codec format definition | Docs GitHub
membrane_mp4_format | MPEG-4 container format definition | Docs GitHub
membrane_opus_format | Opus audio format definition | Docs GitHub
membrane_rtp_format | Real-time Transport Protocol format definition | Docs GitHub
membrane_mpegaudio_format | MPEG audio format definition | Docs GitHub
membrane_h264_format | H264 video format definition | Docs GitHub
membrane_cmaf_format | CMAF definition | Docs GitHub
membrane_matroska_format | Matroska format definition | Docs GitHub
membrane_vp8_format | VP8 Format Description | Docs GitHub
membrane_vp9_format | VP9 Format Description | Docs GitHub

Apps, protocols & plugins' utilities

Package | Description | Links
ex_sdp | Parser and serializer for Session Description Protocol | Docs GitHub
ex_libnice | Libnice-based Interactive Connectivity Establishment (ICE) protocol support for Elixir | Docs GitHub
ex_libsrtp | Elixir bindings for libsrtp | Docs GitHub
ex_dtls | DTLS and DTLS-SRTP handshake library for Elixir, based on OpenSSL | Docs GitHub
membrane_rtsp | RTSP client for Elixir | API Docs GitHub
membrane_common_audiomix | [Experimental] | GitHub
membrane_telemetry_dashboard | Introduction to integrating membrane_timescaledb_reporter repository with membrane_dashboard to monitor your pipeline behaviour | GitHub
membrane_webrtc_server | Signalling server for WebRTC | Docs GitHub

Jellyfish

Jellyfish is a GitHub organization containing Membrane WebRTC-related repositories, focusing on creating a standalone media server.

Plugins

Package | Description | Links
membrane_rtc_engine | RTC Engine and its client library | Docs GitHub
membrane_webrtc_plugin | [Alpha] Plugin for sending and receiving media with WebRTC | Docs GitHub
membrane_ice_plugin | Plugin for ICE protocol | Docs GitHub

Libraries

Package | Description | Links
fake_turn | STUN and TURN library for Erlang / Elixir | GitHub
membrane_rtc_engine_timescaledb | Functions which allow storing Membrane.RTC.Engine metrics reports in a database | Docs GitHub
jellyfish | [In progress] Standalone media server | GitHub

RTC Engine clients

Package | Description | Links
membrane-webrtc-js | JS/TS client library for Membrane RTC Engine | NPM version GitHub
membrane-webrtc-ios | Membrane WebRTC client compatible with Membrane RTC Engine | GitHub
membrane-webrtc-android | Membrane WebRTC client compatible with Membrane RTC Engine | GitHub
react-native-membrane-webrtc | React Native wrapper for membrane-webrtc-ios and membrane-webrtc-android | NPM version GitHub

Book

Package | Description | Links
Jellybook | Theoretical concepts behind general purpose media server. | GitHub

RTC Engine based applications

Package | Description | Links
membrane_videoroom | Video conferencing platform using WebRTC | GitHub
membrane_live | [Alpha] Webinar app in React, Phoenix and Membrane | GitHub

Tutorials

Package | Description | Links
membrane_tutorials | Repository with tutorials text | GitHub
membrane_basic_pipeline_tutorial | Code used in the comprehensive basic pipeline tutorial | GitHub
membrane_videoroom_tutorial | Code used in the videroom tutorial | GitHub
membrane_webrtc_tutorial | Code used in the webrtc tutorial | GitHub

No longer mantained

Package | Description | Links
membrane_libnice_plugin | Replaced by membrane_ice_plugin | Docs GitHub
membrane_element_httpoison | Replaced by membrane_element_hackney | Docs GitHub
membrane_dtls_plugin | DTLS and DTLS-SRTP handshake implementation for Membrane ICE plugin | Docs GitHub
membrane_loggers | Replaced by the Membrane.Logger in Membrane Core | Docs GitHub
membrane_caps_audio_flac | [Suspended] FLAC audio format definition | Docs GitHub
membrane_remote_stream_format | Content of this package has been moved to membrane_core | Docs
membrane_element_rtp_jitter_buffer | Package has become part of membrane_rtp_plugin | Docs
membrane_bin_rtp | Package has become part of membrane_rtp_plugin | Docs
membrane_element_icecast | [Experimental] Element capable of sending a stream into Icecast streaming server | GitHub
membrane_element_live_audiomixer | Simple mixer that combines audio from different sources | GitHub
membrane_element_msdk_h264 | [Experimental] Hardware-accelerated H.264 encoder based on IntelMediaSDK | GitHub
membrane_rtp_aac_plugin | [Alpha] RTP AAC depayloader | Docs GitHub
membrane_element_flac_encoder | [Suspended] | GitHub
membrane_protocol_icecast | [Suspended] | GitHub
membrane_server_icecast | [Suspended] | GitHub

Contributing Guide

General contribution info

Any contributions to Membrane Framework are welcome. If you would like contribute, but you're not sure how to start or have some questions, don't hesitate to contact us via our Discord.

For inquiries related to our commercial services or for other contact information, please visit Membrane website for more information.

When contributing to existing repo:

fork it

apply change on some branch with a meaningful name

create a PR from fork to our repo

await feedback from someone from the team

after passing the review, the PR will be merged <br>

If you wish to create a new plugin you can give us a shout and if it's something we want as a part of our ecosystem we'll create a repo for you, guide you on your work and maintain the plugin in future. However, if feel confident enough to maintain it on your own, you can, of course, create your own repo and hex package - we only ask you to follow the naming conventions of the framework package and modules. In that case, don't forget to let us know about your work so we could include it on the list of available plugins.

Code style guide

Base the package on membrane_template_plugin. Use Elixir formatter, Dialyzer and Credo config specified there.

Indent with 2 spaces, use LF as end of line and utf-8 encoding

For native code, use clang-format with config (.clang-format):

BasedOnStyle: LLVM
IndentWidth: 2

Avoid ambiguous abbreviations (e.g. res) and unnamed ignored variables (e.g. def handle_init(_) -> def handle_init(_options)) [enforced by Credo]

When implementing something defined in some kind of document or standard (e.g. RFC) try to match variable names as closely as possible to the standard and always provide a reference in a comment

Every public module must have @moduledoc and its public functions both @doc and @spec [enforced by Credo]

Modules should have the following layout [enforced by Credo]:

@moduledoc

@behaviour

use

import

alias

require

@module_attribute

defstruct

@type

@callback

@macrocallback

@optional_callbacks

defmacro, defguard, def, etc.

When defining structs, set default values explicitly and omit brackets

  # not preferred
  defstruct [:name, :params, active: true]

  # preferred
  defstruct name: nil, params: nil, active: true

Always use @impl when implementing behaviour. Apply it to every function clause to avoid situations where the last clause has a typo that is not reported as warning since its treated as different function

Read and follow the official documentation writing and library creation guides.

Make sure the generated docs look well. Run mix docs and open doc/index.html in your browser. Configure properly :nest_modules_by_prefix and :groups_for_modules options in mix.exs (see ex_doc docs for reference)

Read and follow elixir style guide. It applies in all the situations not mentioned above.

Naming Conventions

Pure Elixir plugins:

Repo: membrane_X_plugin where X is format/protocol name (e.g. membrane_mpegts_plugin)

Application: :membrane_X_plugin

Modules: Membrane.X.* (e.g. Membrane.AAC.Parser)

Plugins wrapping native library:

Repo: membrane_X_LIB_plugin where X is format/protocol name and LIB - the library (e.g. membrane_aac_fdk_plugin)

Application: :membrane_X_LIB_plugin

Modules: Membrane.X.LIB.* (e.g. Membrane.AAC.FDK.Decoder)

Format definition (former caps):

Repo: membrane_X_format where X is the format name (e.g. membrane_aac_format)

Application: :membrane_X_format

Format struct: %Membrane.X{} (e.g. %Membrane.AAC{channels: 2})

Other modules: Membrane.X.*, the same namespace as plugins for this format (e.g. Membrane.AAC.*)

Contributing practices

When creating contribution:

You can open a draft PR right away.

If PR solves GH issue, remember to link the issue to the PR (either by adding Closes # and PR number or using the UI)

When finished, make a self-review before assigning a reviewer.

Remember to watch for comments accidentally marked as outdated and don't mark someone else's comments as resolved.

When you consider all the comments fixed, re-request the review.

Do not force push changes to the reviewed commits - this breaks changes since last review functionality on GH

Always remove your branch after merging the PR. It should happen automatically. If not, ask someone with permissions to enable that in settings.

Elements

Elements in Membrane Framework are the most basic entities responsible for processing multimedia. Each instance of an element is an Erlang process, that has an internal state and communicates by message passing.

Every element is created to solve one problem. The examples of elements are audio encoder (converting raw audio to encoded format like MP3), file source (reading data from a file and passing it to other elements) or UDP sink (sending data from an application via UDP socket).

Element types

There are three basic types of elements:

Source - responsible for getting (or generating) data delivering it to other elements

Sink - defines an endpoint for data flowing through an application.

Filter - an element that receives data from other elements, processes it and sends it further to the next elements

Playback states

Every element may be in one of three states:

:stopped - element has been created

:prepared - element should be ready for processing data. All necessary resources should be allocated and initialized.

:playing - element is actually processing data

Element's lifecycle

Proper resource management is crucial when creating reliable and stable software. For this purpose, Membrane Framework introduces few rules that unify expected behaviour of elements, so it is possible to manage them better (i.e. in pipelines). The main tool for this is a playback state. Elements should react to every state change and know exactly what should be done at the specific time of their "life".

Callback handle_init

Above callback is invoked only once, upon the element creation. It receives options specified by the user, which should be parsed and on their base, the element should create and initialize its internal state.

State stopped

It is the first and the last state of every element. While being in it, elements should not have any "dynamic" resources, opened files or devices. Elements in the stopped state might still not be linked, thus they shouldn't return from callbacks any actions that require sending any message via pads (like caps and events).

Stopped -> prepared change

It is a good place to allocate all needed resources (for example native resources). Elements in pipelines are already linked when performing this callback. If it is possible to know the capabilities of processed streams, they should be sent to other elements via :caps action.

State prepared

Elements should have already all needed resources allocated in order to process data. Although all timers should be still stopped and waiting for the stream start. Elements are permitted to send caps and events via their pads, but sticky events will be queued in a pull buffer and processed after state change to the playing. Nevertheless, other events (not sticky) will be handled with handle_event.

From this state, it is possible to transit to both 'stopped' and 'playing' states. Elements should be prepared for both of these changes and don't assume any of them.

Prepared -> playing change

If an element is based on a timer, it should be started in handle_prepared_to_playing callback. Also, most Sinks should make their first demand in this callback.

State playing

Elements are processing data, timers are running. Sending buffers and demands is allowed now.

Playing -> prepared change

In callback handling this change, the element should stop all running timers. Since this moment, elements won't process any more buffers or sticky events. All the data waiting in InputBuffers will be dropped.

Prepared -> stopped change

When handling this change, all allocated resources should be cleaned, devices closed. The internal state of an element should be ready for the next transition to prepared state.

Pads and capabilities

To create the flow of data between elements in the application, they have to communicate with each other. For that purpose, the concept of pads and capabilities is used. Element's pads, much like contact pads on printed circuit board, are inputs and outputs of an element and are used to connect the elements with one another. Because of that, there are two types of pads: input and output. It is worth mentioning that Source elements may only contain output pads, Sinks contain only input pads, and Filters can have both of them.

Every pad has some capabilities, that define a type of data that it is expecting. This format can be, for example, raw audio with a specific sample rate or encoded audio in a given format.

In order to send data between elements, their pads need to be linked. There are a couple of rules that apply to pad linking:

One pad of an element can only be linked with one pad from another element. (dynamic pads can help with that limitation)

Only links between output and input pads are allowed.

Capabilities of pads have to be compatible.

Properties

Each pad has properties that can be set when defining a pad:

Availability - either :always - meaning the pad is static and available from the moment an element is spawned or :on_request meaning it is dynamic

Mode:

:push - it's a simple mode where an element producing data pushes it right away on the output pad and input pad in this mode should be always ready to process that data

:pull - this mode provides a back-pressure mechanism - Source/Filter is only allowed to send data after receiving a demand from a downstream element and the responsibility of Sink/Filter is to send those demands

Demand unit (only pull input pads) - specifies what unit will be used to send demands (either :bytes or :buffers)

Caps - capabilities supported by the pad

Options - specification of options accepted by the pad

Dynamic Pads

A dynamic pad is a type of pad that acts as a template - each time some other pad is linked to a dynamic pad a new instance of it is created.

Why do we need dynamic pads?

Dynamic pads don't have to be linked when the element is started. Obviously, the element has to support that, but in return, it gives new possibilities when the number of pads can change on-the-fly.

Another use case for dynamic pads is when specifying pads manually becomes cumbersome. Let's consider an audio mixer - it is a type of element that is likely to have numerous input pads with the same definition. Thanks to dynamic pads there's no need to copy-paste the same definition over and over again. Plus, you won't be limited by the number of pads that have been defined.

Creating an element with dynamic pads

Creating an element with dynamic pads is not much different than creating one with static pads. The key difference is that we need to specify that one of the pads is dynamic, by setting pad availability to :on_request.

Now, each time some element is linked to this pad, a new instance of the pad is created and callback Membrane.Element.Base.handle_pad_added/3 is invoked. Instances of a pad can be referenced as Pad.ref(pad_name, pad_id).

Gotchas

As usual, with great power comes great responsibility. When implementing an element with dynamic pads you need to remember to implement Membrane.Element.Base.handle_pad_added/3 and Membrane.Element.Base.handle_pad_removed/3 callbacks. Membrane.Element.Base.handle_event/4 might also need some attention as the default implementation won't support dynamic pads. And of course, the logic of an element may become more complicated as it has to support changing number of pads.

Options

Both elements and their pads may define their own options that parametrize their work. For example, some audio decoder may have an option named bitrate that represents bitrate of the output data.

The options for an element are passed when the element is created, while pad options are provided when two elements are linked

Defining elements

Elements are Elixir modules that use a proper module (either Membrane.Sink, Membrane.Filter or Membrane.Source). They have to define options and pads using provided macros (Membrane.Element.WithInputPads.def_input_pad/2, Membrane.Element.WithOutputPads.def_output_pad/2 and Membrane.Element.Base.def_options/1) and implement at least required callbacks.

Pipelines

Conceptually, a pipeline is a container that consists of many elements and links between them. Like an element, every pipeline has a playback state (and uses the same callbacks to handle transitions between them).

Implementation-wise, it's an Erlang process responsible for starting elements (usually referred to as children). It also manages their playback state to match the one set for the pipeline.

Communication with children

During the application execution, elements may want to signal some events. For that purpose, they send the Notification to their watcher, which in most cases is a pipeline.

A pipeline can also be used as a gateway to an element - as elements' pids aren't available globally pipeline can route messages to children using names defined at pipeline creation.

Defining pipelines

Pipelines are Elixir modules that use Membrane.Pipeline and implement callbacks required by Membrane.Pipeline behaviour.

Bins

Bins, similarly to pipelines, are containers for elements. However, at the same time, they can be placed and linked within pipelines. Although bin is a separate Membrane entity, it can be perceived as a pipeline within an element. Bins can also be nested within one another.

There are two main reasons why bins are useful:

they enable creating reusable element groups

they allow managing their children, for instance by dynamically spawning or replacing them as the stream changes.

Pads

Bin's pads are defined similarly to element's pads and can be linked in similar way. However, their role is limited to proxy the stream to elements and bins inside (inputs) or outside (outputs). To achieve that, each input pad of a bin needs to be linked to both an output pad from the outside of a bin and an input pad of its child inside. Accordingly, each bin's output should be linked to output inside and input outside of the bin.

Bins and the stream

Although the bin passes the stream through its pads, it does not access it directly, so that callbacks such as handle_process or handle_event are not found there. This is because the responsibility of the bin is to manage its children, not to process the stream. Whatever the bin needs to know about the stream, it should get via notifications from the children.

Bins as black boxes

Bins are designed to take as much responsibility for their children as possible so that pipelines (or parent bins) don't have to depend on them. That's why notifications from the children are sent to their parent bin only (or to the pipeline in case of top-level elements or bins). Also, messages received by a bin or pipeline can be forwarded only to its direct children.

Creating the first pipeline

In this chapter, we will implement an example of using elements and grouping them into the pipeline. Namely, we will write an application that reads .mp3 file and using PortAudio library plays its content to the default audio device in your system.

Source code for this pipeline as well as an mp3 file sample can be found in membrane demo repository.

Add dependencies to mix.exs

Membrane Framework is modular and consists of many packages available on hex.pm To start the work, we have to add a dependency to our main package - Membrane Core, which contains all mechanisms used for creating and managing pipelines and elements. To do this, just add the following line to the deps in your mix.exs:

{:membrane_core, "~> 0.7.0"},

Furthermore, there are quite a few Membrane elements providing different functionalities and supporting a variety of multimedia formats. Each element is available as a separate package.

In this tutorial, we will use Membrane.Element.File (for reading data from a file), Membrane.Element.FFmpeg.SWResample.Converter (for audio format conversion) and Membrane.Element.PortAudio (for writing the audio to audio device):

{:membrane_file_plugin, "~> 0.6.0"},
{:membrane_portaudio_plugin, "~> 0.8.0"},
{:membrane_ffmpeg_swresample_plugin, "~> 0.8.0"},
{:membrane_mp3_mad_plugin, "~> 0.8.0"}

These dependencies rely on native libraries that have to be available in your system. You can use the membraneframeworklabs/docker_membrane docker image or the following commands to install them.

MacOS

brew install mad ffmpeg portaudio pkg-config

Ubuntu

sudo apt-get install libmad0-dev libswresample-dev libavutil-dev portaudio19-dev

Arch / Manjaro

sudo pacman -S ffmpeg libmad portaudio pkg-config

Create a module for our pipeline

To define a pipeline you have to create an empty module and add use Membrane.Pipeline clause.

defmodule Your.Module.Pipeline do
  use Membrane.Pipeline

  ...

Add handle_init definition

Elements used in the pipeline and links between them should be given in handle_init function. This function receives a single argument - configuration/options, which are given when the pipeline is started. In our case, it will be a string containing the path to the .mp3 file to play.

@impl true
def handle_init(path_to_mp3) do
  ...
end

Inside handle_init, we should define all elements and links between them. Firstly, let's create the keyword list, that contains all elements that will be used in our application. Key of the keyword list represents the name that we give to the element. Value is an element specification.

  children = %{
    file: %Membrane.File.Source{location: path_to_mp3},
    decoder: Membrane.MP3.MAD.Decoder,
    converter: %Membrane.FFmpeg.SWResample.Converter{output_caps: %Membrane.Caps.Audio.Raw{sample_rate: 48_000, format: :s16le, channels: 2}},
    player: Membrane.PortAudio.Sink,
  }

Notice, that there are two approaches to element declarations: as a module name or as a struct of given module. The second approach gives the possibility to pass some additional argument.

Then, we should specify links using dedicated DSL:

  links = [
    link(:file)
    |> via_out(:output)
    |> via_in(:input)
    |> to(:decoder)
    |> via_out(:output)
    |> via_in(:input)
    |> to(:converter)
    |> via_out(:output)
    |> via_in(:input)
    |> to(:player)
  ]

Since used elements define pads with default names - :output and :input, we can skip via_out and via_in parts:

  links = [
    link(:file) |> to(:decoder) |> to(:converter) |> to(:player)
  ]

Last but not least, we should return created terms in the correct format - Membrane.ParentSpec struct. Note that it's aliased by default by Membrane.Pipeline.__using__/1.

  spec = %ParentSpec{
    children: children,
    links: links
  }

  {{:ok, spec: spec}, %{}}

The return value contains also an empty map. It is a new state for the pipeline, which gives a possibility to store some additional information for later use. In this case, it is unnecessary.

To sum up, the whole file can look like this:

defmodule Your.Module.Pipeline do
  use Membrane.Pipeline

  @impl true
  def handle_init(path_to_mp3) do
    children = %{
      file: %Membrane.File.Source{location: path_to_mp3},
      decoder: Membrane.MP3.MAD.Decoder,
      converter: %Membrane.FFmpeg.SWResample.Converter{
        output_caps: %Membrane.Caps.Audio.Raw{
          sample_rate: 48_000,
          format: :s16le,
          channels: 2
        }
      },
      player: Membrane.PortAudio.Sink,
    }

    links = [
      link(:file) |> to(:decoder) |> to(:converter) |> to(:player)
    ]

    spec = %ParentSpec{
      children: children,
      links: links
    }

    {{:ok, spec: spec}, %{}}
  end

end

Run the pipeline

The simplest way to create and run above pipeline is to type in iex console:

alias Your.Module.Pipeline
{:ok, pid} = Pipeline.start_link("/path/to/mp3")
Pipeline.play(pid)

The given .mp3 file should be played on the default device in your system. Please use .mp3 that has no ID3 or ID3v2 tags.

The demo available here contains an .mp3 file without tags.

More advanced pipelines

After successfully running your first Membrane Pipeline, we can move on to more complicated stuff. In this guide, we will show you how to provide options for pads, how to link an element with dynamic pad(s) and how to introduce synchronization of start between elements.

Pad options

Not only elements can have options. Some pads have options as well. You can provide them in a keyword list within the second argument of Membrane.ParentSpec.via_out/2 or Membrane.ParentSpec.via_in/2:

links = [
  # ...
  link(:decoder) |> via_in(:input, options: [mute: true]) |> to(:mixer),
  # ...
]

Available pad options are documented in every element's main module in automatically generated Pads section.

When an input pad works in :pull mode you can also configure its input buffer:

links = [
  # ...
  link(:decoder) |> via_in(:input, buffer: [preffered_size: 42_000]) |> to(:mixer),
  # ...
]

Available settings are described in the Membrane.Core.InputBuffer.props_t/0 docs.

Of course, buffer and pad options can be combined:

links = [
  # ...
  link(:decoder) |> via_in(:input, options: [mute: true], buffer: [preffered_size: 42_000]) |> to(:mixer),
  # ...
]

Dynamic pads

Dynamic pads (described here) can be linked just like any other pad. The main difference is that each link creates a new instance of this pad, so it can be linked multiple times.

links = [
  # ...
  link(:decoder_a) |> to(:mixer),
  link(:decoder_b) |> to(:mixer),
  link(:decoder_c) |> to(:mixer),
  # ...
]

You can also explicitly specify the reference of a dynamic pad that will be used. To create such reference, use Membrane.Pad.ref/2:

links = [
  # ...
  link(:decoder_a) |> via_in(Pad.ref(:input, :a)) |> to(:mixer),
  link(:decoder_b) |> via_in(Pad.ref(:input, :b)) |> to(:mixer),
  link(:microphone) |> via_in(Pad.ref(:input, 1)) |> to(:mixer),
  # ...
]

Here's an example of a pipeline using an element with a dynamic output pad - Membrane.Element.Tee.Master:

defmodule MultipleCopyPipeline do
  use Membrane.Pipeline
  alias Membrane.Element.{File, Tee}

  @impl true
  def handle_init(_) do
    children = [
      file_src: %File.Source{location: "/tmp/source_file"},
      tee: Tee.Master,
      file_sink1: %File.Sink{location: "/tmp/destination_file1"},
      file_sink2: %File.Sink{location: "/tmp/destination_file2"},
      file_sink3: %File.Sink{location: "/tmp/destination_file3"}
    ]

    links = [
      link(:file_src) |> to(:tee),
      link(:tee) |> via_out(:master) |> to(:file_sink1),
      link(:tee) |> via_out(:copy) |> to(:file_sink2),
      link(:tee) |> via_out(:copy) |> to(:file_sink3)
    ]

    state = %{}

    {{:ok, spec: %ParentSpec{children: children, links: links}}, state}
  end
end

This example requires the following dependencies:

  defp deps do
    [
      {:membrane_element_tee, "~> 0.5.0"},
      {:membrane_file_plugin, "~> 0.6.0"}
    ]
  end

Selecting a clock

Some elements (most likely sinks) may provide its clock that can be used by the elements in the pipeline to create timers generating ticks and informing about passing time. For example a sink, sending audio to a sound card may provide a clock based on a hardware clock on the device.

If the pipeline ignores that clock, the clocks are not aligned, i.e. time is passing slower according to one of them, and one of the elements inside the pipeline produces data according to VM time, the tempo of data generation will be different from the tempo of data consumption. This will eventually result in either buffer overflow or underflow (causing, for example, audible 'cracks' in audio playback if the proper amount of audio samples is not available on time).

Clock provider is an element that exports clock that should be used as the pipeline clock - the default clock used by elements' timers. When there is only one element providing clock, the pipeline can choose it automatically. When there are two or more such elements, you can set it by providing an atom with element's name via :clock_provider field inside Membrane.ParentSpec struct:

%ParentSpec{
  children: [
    # ...
    hardware_sink: # ... ,
    # ...
  ]
  # ...
  clock_provider: :hardware_sink,
  # ...
}

Synchronization

Sometimes, you may need to synchronize some of the elements within a pipeline. A good example of a situation where such synchronization is needed is playing audio and video with 2 separate sinks. To do this you can use :stream_sync field in Membrane.ParentSpec struct to specify elements that should start playing at the same moment. You can set it to :sinks atom synchronizing all sinks in the pipeline:

%ParentSpec{
  # ...
  stream_sync: :sinks
  # ...
}

or a list of groups (lists) of elements synchronizing all elements in each group:

%ParentSpec{
  children: [
    # ...
    element1: # ... ,
    element2: # ... ,
    element3: # ... ,
    element4: # ... ,
    # ...
  ]
  # ...
  stream_sync: [[:element1, :element2], [:element3, :element4]]
  # ...
}

It is worth mentioning that to keep the stream synchronized all involved elements need to rely on the same clock.

By default, no elements are synchronized.

Logging

Apart from usual Logger configuration, logging in Membrane can be additionally configured, also via Elixir's Config. It allows to enable verbose mode and customize metadata, for example:

config :membrane_core, :logger, verbose: true

See Membrane.Logger for details.

Moreover, pipelines support Membrane.Pipeline.Action.log_metadata_t/0, that enables setting logger metadata to all descendants of a pipeline, for example:

@impl true
def handle_init(opts) do
  # ...
  {{:ok, log_metadata: [pipeline_id: opts.id]}, state}
end

To have the metadata displayed, remember to enable that in the logger backend, for example:

config :logger, :console, metadata: [:pipeline_id]

The log_metadata action is also available in bins: Membrane.Bin.Action.log_metadata_t/0.

Implementing first element

Implementing a new element is very similar to declaring a new pipeline. All you have to do is create a new module and implement some callbacks.

The first decision to make is to specify what kind of element we are going to implement: source, sink or filter. In this chapter, we will implement a very simple filter that counts received buffers and passes them to the next element. It will build simple statistics and send them to Pipeline via Membrane.Notification mechanism.

Source code for this filter can be found in the membrane demo repository.

Base module

To indicate the choice of implementing 'filter' we have to add the following line to our module:

use Membrane.Filter

This macro forces us to invoke some macros or implement some methods in our module:

def_options macro - macro that defines known options for the element type. It automatically generates the appropriate struct.

def_input_pad and def_output_pad - macros that define the pads of the element

callbacks handle_init, handle_demand and handle_process - callbacks that are invoked when initializing the element, handling incoming demand or buffer

Options

Our element will have only one option - time interval, telling how often statistics should be sent and zeroed. Its definition will look like the following:

def_options interval: [
    type: :integer,
    default: 1000,
    description: "Amount of the time in milliseconds, telling how often statistics should be sent and zeroed"
  ]

Pads

It is very common for the filter to declare two pads that are called :input and :output

def_input_pad :input,
  availability: :always,
  mode: :pull,
  demand_unit: :bytes,
  caps: :any,
  options: [
    divisor: [
      type: :integer,
      default: 1,
      description: "Number by which the counter will be divided before sending notification"
    ]
  ]

def_output_pad :output,
  availability: :always,
  mode: :pull,
  caps: :any

In above definition, availability :always means that pad of this element is always available. :pull mode means that this element sends buffers to the next element only when they are demanded. When a demand is received on an output pad, the handle_demand callback is invoked with total demand for the pad and a unit - either :bytes or :buffers.

For input pads in :pull mode one more entry has to be provided - :demand_unit. It determines in which unit the demand is sent to the upstream element and can be set, as mentioned, to :bytes or :buffers.

The next element in the keyword list represents the capabilities (caps) of the pad. :any means that any type of buffer can be passed on this pad. If you want to restrict the types of data allowed on this pad you can define caps specifications as described in docs for module Membrane.Caps.Matcher.

Last entry for input pad defines an option for this pad added just for demonstration purpose. The options for pads are defined just like element's options in def_options macro.

handle_init/1

In handle_init we initialize the internal state of the element. As the first argument of callback, options will be received. In our state, we will create variable for counting buffers and timer. Timer won't be initialized at this point, we will wait for handle_prepared_to_playing/2 callback, which informs that pipeline is in the :playing state.

@impl true
def handle_init(%__MODULE__{interval: interval}) do
  state = %{
    interval: interval,
    counter: 0,
    timer: nil
  }
  {:ok, state}
end

handle_prepared_to_playing/2

Now it is a time to start the timer that will send :tick messages to our element when it should flush results and reset the counter. We should also store the created timer, to be able to release it when the pipeline will stop processing data (in handle_prepared_to_stopped/2).

@impl true
def handle_prepared_to_playing(_ctx, state) do
  {:ok, timer} = :timer.send_interval(state.interval, :tick)
  {:ok, %{state | timer: timer}}
end

handle_demand/5

Since both of our pads work in :pull mode, we have to handle incoming demands. In our case this task is very simple - we have to just redirect incoming demands to the previous element.

For that purpose, we will return an action as an additional term in the output tuple: {{:ok, action_list}, new_state}. Actions are generally speaking the activity that we request the element to perform. The example actions are: sending buffer/event/new_caps on some pad, sending messages to the pipeline or - like in our case - sending demand through some input pad.

Actions are always the entries in the keyword list, where the key is an atom indicating the action name and the value contains the parameters of the action. In this case, it is a tuple with the pad name at first position and size of the demand on the second position.

More information on how the demands work can be found in a separate chapter

@impl true
def handle_demand(:output, size, :bytes, _context, state) do
  {{:ok, demand: {:input, size}}, state}
end

handle_process/4

Incoming buffers are processed in this callback. We will update our counter and pass the buffer to the :output pad.

Overriding handle_process callback means that we want to receive only one buffer at a time. However, keep in mind that very often buffers are delivered to the element in groups. It is also possible to override handle_process_list callback, which by default performs action :split that in turn invokes handle_process callback for each buffer from the group.

@impl true
def handle_process(:input, %Membrane.Buffer{} = buffer, _context, state) do
  new_state = %{state | counter: state.counter + 1}
  {{:ok, buffer: {:output, buffer}}, new_state}
end

handle_other/3

All messages sent to the element's process that were not recognized as internal membrane messages (like buffers, caps, events or notifications) are handled in handle_other/3. We will receive our ticks here, so we will know that we should zero the counter. It is also a good place to send a notification to the pipeline with the statistics.

This also presents how pad options can be accessed via context.

@impl true
def handle_other(:tick, ctx, state) do
  # create the term to send
  notification = {
    :counter,
    div(state.counter, ctx.pads.input.options.divisor)
  }

  # reset the timer
  new_state = %{state | counter: 0}

  {{:ok, notify: notification}, new_state}
end

handle_prepared_to_stopped/2

Last but not least there is the handle_prepared_to_stopped/2 callback. It is the place to stop the timer and clean up.

@impl true
def handle_prepared_to_stopped(_ctx, state) do
  {:ok, :cancel} = :timer.cancel(state.timer)
  {:ok, %{state | counter: 0, timer: nil}}
end

Summary

The complete code of our element can look like this:

defmodule Your.Module.Element do
  use Membrane.Element.Base.Filter

  def_options interval: [
                type: :integer,
                default: 1000,
                description:
                  "Amount of the time in milliseconds, telling how often statistics should be sent and zeroed"
              ]

  def_input_pad :input,
    availability: :always,
    mode: :pull,
    demand_unit: :bytes,
    caps: :any,
    options: [
      divisor: [
        type: :integer,
        default: 1,
        description: "Number by which the counter will be divided before sending notification"
      ]
    ]

  def_output_pad :output,
    availability: :always,
    mode: :pull,
    caps: :any

  @impl true
  def handle_init(%__MODULE{interval: interval}) do
    state = %{
      interval: interval,
      counter: 0,
      timer: nil
    }

    {:ok, state}
  end

  @impl true
  def handle_prepared_to_stopped(_ctx, state) do
    {:ok, :cancel} = :timer.cancel(state.timer)
    {:ok, %{state | counter: 0, timer: nil}}
  end

  @impl true
  def handle_prepared_to_playing(_ctx, state) do
    {:ok, timer} = :timer.send_interval(state.interval, :tick)
    {:ok, %{state | timer: timer}}
  end

  @impl true
  def handle_demand(:output, size, :bytes, _context, state) do
    {{:ok, demand: {:input, size}}, state}
  end

  @impl true
  def handle_process(:input, %Membrane.Buffer{} = buffer, _context, state) do
    new_state = %{state | counter: state.counter + 1}
    {{:ok, buffer: {:output, buffer}}, new_state}
  end

  @impl true
  def handle_other(:tick, ctx, state) do
    # create the term to send
    notification = {
      :counter,
      div(state.counter, ctx.pads.input.options.divisor)
    }

    # reset the timer
    new_state = %{state | counter: 0}

    {{:ok, notify: notification}, new_state}
  end
end

Testing

We strongly encourage contributors to test the elements they create. We usually advise on creating two types of tests: Unit Tests and Integration Tests. To facilitate writing tests and make the developer experience as smooth as possible, we expose a couple of testing utilities inside Membrane.Testing package.

We will discuss and walk you through writing tests on a real-world example for the AAC decoder element.

Setting up

In order to simplify things and make the test easier to maintain and modify, we propose the following solution:

Create a sample file for the element under test. As an example, for the AAC decoder we'll use a very simple AAC audio file

Create a known output as a reference file. For example, we save raw audio frames as the reference output for our AAC Decoder. Note: This may not always be the best solution since most encoders produce a non-deterministic output for the same input

Use the input file and reference file as sources of truth for your tests

Running tests

To run all tests, execute:

mix test

Unit Tests

We will create unit tests for our Native decoder.

We start with a standard test module definition

defmodule Decoder.NativeTest do
  use ExUnit.Case
  alias Membrane.Element.AAC.Decoder.Native

Our test scenario will check if the native decoder is able to decode a single frame of data:

  test "Decode 1 AAC frame" do

As described above, we will use input/reference files as our data sources. Those files are placed inside test/fixtures directory

    in_path = "fixtures/input-sample.aac" |> Path.expand(__DIR__)
    reference_path = "fixtures/reference-sample.raw" |> Path.expand(__DIR__)
    ...
    {:ok, file} = File.read(in_path)

The first step is to ensure we can create a Native decoder:

    assert {:ok, decoder_ref} = Native.create()

Next, we will retrieve the first frame from the input file, pass it to the Native decoder and assert that it was decoded correctly. Subsequent calls to decode_frame should return :not_enough_bits, as all data is already parsed.

    assert <<frame::bytes-size(256), _::binary>> = file
    assert :ok = Native.fill(frame, decoder_ref)
    assert {:ok, decoded_frame} = Native.decode_frame(frame, decoder_ref)
    assert {:error, :not_enough_bits} = Native.decode_frame(frame, decoder_ref)

Finally, we'll compare the decoded frame with a reference frame from the saved raw file.

    {:ok, ref_file} = File.read(reference_path)

    assert <<ref_frame::bytes-size(4096), _::binary>> = ref_file
    ...
    assert bit_size(decoded_frame) == bit_size(ref_frame)
    assert Membrane.Payload.to_binary(decoded_frame) == ref_frame
  end
end

Integration tests

For integration tests, we'll check if the whole decoding pipeline with our AAC decoder works correctly.

Our test pipeline uses Membrane.Testing.Pipeline module. This means that we only need to specify pipeline's Membrane.Testing.Pipeline.Options and all elements' links and callbacks are automatically implemented for us.

  Pipeline.start_link(%Pipeline.Options{
    elements: [
      file_src: %Membrane.Element.File.Source{location: in_path},
      decoder: Membrane.Element.AAC.Decoder,
      sink: %Membrane.Element.File.Sink{location: out_path}
    ]
  })

Now, onto our test case module:

defmodule DecoderTest do
  use ExUnit.Case

  import Membrane.Testing.Assertions

  alias Membrane.Pipeline

First, we define a couple of helper methods.

prepare_paths generates file paths for our input/reference files as well as creates a temporary output file for our testing pipeline which will automatically be cleaned up after the test is finished:

  def prepare_paths(filename) do
     in_path = "fixtures/input-#{filename}.aac" |> Path.expand(__DIR__)
     reference_path = "fixtures/reference-#{filename}.raw" |> Path.expand(__DIR__)
     out_path = "/tmp/output-decoding-#{filename}.raw"
     File.rm(out_path)
     on_exit(fn -> File.rm(out_path) end)
    {in_path, reference_path, out_path}
  end

assert_files_equal compares two on-disk files:

  def assert_files_equal(file_a, file_b) do
    assert {:ok, a} = File.read(file_a)
    assert {:ok, b} = File.read(file_b)
    assert a == b
  end

Finally, our test case. Notice the Membrane.Testing.Assertions.assert_end_of_stream/4 call which waits and validates that the EndOfStream message was received.

  describe "Decoding Pipeline should" do
    test "Decode AAC file" do
      {in_path, reference_path, out_path} = prepare_paths("sample")
      assert {:ok, pid} =  Pipeline.start_link(%Pipeline.Options{
        elements: [
          file_src: %Membrane.Element.File.Source{location: in_path},
          decoder: Membrane.Element.AAC.Decoder,
          sink: %Membrane.Element.File.Sink{location: out_path}
        ]
      })

      assert Pipeline.play(pid) == :ok # Start the pipeline
      assert_end_of_stream pid, :sink
      assert_files_equal(out_path, reference_path) # Compare pipeline output with reference file
    end
  end
end

Once again, we are comparing the pipeline's output with our reference file to assert that everything works correctly.

Customizing pipeline module

Membrane.Testing.Pipeline allows us to provide our custom implementations of Membrane.Pipeline callbacks that will be executed prior to Membrane.Testing.Pipeline's implementations. Let's create pipeline that will answer to requests with response we passed in initialization options. To do this, we must define module which implements callbacks we want.

defmodule Example.Pipeline do
  use Membrane.Pipeline

  @impl true
  def handle_init(options) do
    state = %{response: options.response}
    # put spec of choice into spec
    {{:ok, spec}, state}
  end

  @impl true
  def handle_other({:request, from}, state) do
    send(from, {:response, state.response})
    {:ok, state}
  end

  @impl true
  def handle_other(_message, state),
    do: {:ok, state}
end

In order to use callbacks defined in Example.Pipeline and pass custom initialization arguments, we have to specify it in Options. Please take note that there is no elements nor links options provided, because they can't be used with module override and will result in error when Membrane.Testing.Pipeline.start_link/2 or Membrane.Testing.Pipeline.start/2 is called.

Pipeline.start_link(%Pipeline.Options{
  module: Example.Pipeline,
  custom_args: %{response: "Hello there!"}
})

Now we can create another test case to try out new functionality.

describe "Decoding Pipeline should" do
  ...

  test 'Answer with {:reponse, "Hello there!"} message' do
    assert {:ok, pid} =  Pipeline.start_link(%Pipeline.Options{...})
    assert Pipeline.play(pid) == :ok # Start the pipeline
    test_process_pid = self()
    send(pid, {:request, test_process_pid})
    assert_pipeline_receive(pid, {:request, test_process_pid}) # Check if pipeline got message
    assert_receive({:response, "Hello there!"}) # Check if we got return message
  end
end

Using testing elements

Apart from the Membrane.Testing.Pipeline, which we've already seen, there are a bunch of other testing utilities which may come in handy for different test scenarios:

Membrane.Testing.Source - Can be either used as an alternative for File.Source allowing you to pass in a list of payloads that will be supplied to the pipeline or it will output data based on the actions_generator. It can be useful for generating sequential payloads or a random input.

Membrane.Testing.Sink - A fake sink element that will pass all received buffers, events and caps to parent pipeline. Useful for asserting output buffers one by one.

We also prepared a bunch of assertions that work in conjunction with both Membrane.Testing.Pipeline and said elements. They are stored in Membrane.Testing.Assertions, they usually come in two flavors: assert_* and refute_* similarly to ExUnit assertions.

Testing sink

Membrane.Testing.Sink reports caps, events and buffers it receives to its pipeline. If you are using it within Membrane.Testing.Pipeline you can make assertions about what the Membrane.Testing.Sink has received. For example, this is how Membrane.Element.Tee.Parallel could be tested:

alias Membrane.Testing.{Source, Pipeline, Sink}

range = 1..100
{:ok, pipeline} =
    Pipeline.start_link(%Pipeline.Options{
      elements: [
        src: %Source{output: range},
        tee: Tee,
        sink1: %Sink{},
        sink2: %Sink{},
      ],
      links: [
        link(:src)
        |> to(:tee)
        |> to(:sink1),
        link(:tee)
        |> to(:sink2)
      ]
    })

Membrane.Pipeline.play(pipeline)

# assert every message was received three times
Enum.each(range, fn payload ->
  assert_sink_buffer(pid, :sink1, %Buffer{payload: ^payload})
  assert_sink_buffer(pid, :sink2, %Buffer{payload: ^payload})
end)

# Wait for EndOfStream message on every sink
assert_end_of_stream(pid, :sink1, :input, 3000)
assert_end_of_stream(pid, :sink2, :input, 3000)

Testing element's interaction with a pipeline

Another common use of these assertions is checking whether communication with a pipeline is proceeding correctly. You can either check whether Pipeline received a message that would be handled by c:Membrane.Parent.handle_other/2:

assert_pipeline_receive(pipeline_pid, {:topic, _})

Such an assertion would assert that message matching pattern {:topic, _} will be sent to pipeline with pid pipeline_pid.

If you were, for example, to implement a demuxer and you want to ensure pipeline is notified by the demuxer asking for mapping of streams to pads, you would do it like this:

assert_pipeline_notified(pipeline, :tested_demuxer, {:mapping, mapping})
assert mapping == expected_mapping

Summary

Full source code for the above examples can be found in membrane-element-aac and membrane-element-tee repositories.

Demands

When the link between two Membrane elements works in :pull mode, data between them is sent only when requested by an element consuming buffers. That "request" in our framework is called "demand".

Demand unit

Demands might have one of the following units:

:bytes — self-explanatory, refers to the number of bytes as an actual amount of data

:buffers — when demanding n buffers, element expects to receive n Membrane.Buffer structs. Their size might be dependent on capabilities, the previous elements or their configuration. For example, element Membrane.Element.File.Source has an option named chunk_size that defines how big the buffers are.

Demand units handled by element should be specified as parameters of def_input_pads, i.e.:

def_input_pad :input,
  mode: :pull,
  demand_in: :bytes,
  ...

Handling incoming demand in a downstream element

When an element that produces buffers (Source or Filter) receives a demand, it is expected to produce data and send it through one of its output pads. Elements handle incoming demands via handle_demand/5 callback that is supplied with the following parameters:

pad - the name of the output pad on which the demand has been received

size - the amount of data that is expected to be sent on given pad. It is important to note that the entire demand is always passed to handle_demand and it overrides previous value.

unit - unit of the demand, self-explanatory

context - Membrane.Element.CallbackContext.Demand structure, contains useful information like actual playback state of the element or size of the last demand (in a case where this size is equal to 0 it means that handle_demand has been triggered by :redemand action, see below)

state - actual internal state of the element (like in every other callback)

Below, the very simple case of handling demand in Source element is presented. It just produces buffers on the spot in the handle_demand callback:

@impl true
def handle_demand(:output, size, :buffers, _ctx, state) do
  buffers =
    1..size |> Enum.map(fn num -> %Membrane.Buffer{payload: generate_payload(num)} end)

  {{:ok, buffer: {:output, buffers}}, state}
end

Demanding data from Sink or Filter

Demanding is done by returning {:demand, {pad_name, demand_size}} action from one of the callbacks. For Filters the most common case is to return :demand from handle_demand callback. It can be perceived as translating received demand from output pad to the other upstream element through the input pad:

@impl true
def handle_demand(:output, size, _unit, %Ctx.Demand{}, state) do
  {{:ok, demand: {:input, size}}, state}
end

Like when receiving demand, passed value is meant to override the previous one.

If there is a need to refer to the previous value, the anonymous function can be provided, i.e.:

@impl true
def handle_event(:output, %SomeEvent{}, _context, state) do
  {{:ok, demand: {:input, & &1+1}}, state}
end

Above example just increments by one more buffer (or byte, depending on the demand_unit) the value that was demanded previously.

Action :redemand

Generally, :redemand action can be used in Sources or Filters. Its usage differs little in case of these two elements types, but the effect is the same: invoking handle_demand callback again. When returning :redemand action from the callback, it should be supplied with the name of the appropriate output pad, i.e.:

@impl true
def handle_other(%Membrane.Buffer{} = buffer, _ctx, state) do
  {{:ok, buffer: {:output, :buffer}, redemand: :output}, state}
end

:redemand in Sources

In case of Sources, :redemand is just a helper that simplifies element's code. The element doesn't need to generate the whole demand synchronously at handle_demand or store current demand size in its state, but it can just generate one buffer and return :redemand action. If there is still one or more buffers to produce, returning :redemand will trigger the next invocation of handle_demand Element will produce next buffer and call :redemand again. If there are no more buffers demanded, handle_demand won't be invoked and the loop will end. One more advantage of the approach with :redemand action is that produced buffers will be sent one after another in separate messages and this could possibly improve the latency.

:redemand in Filters

:redemand in Filters is useful in a situation where not the entire demand of output pad has been satisfied and there is a need to send a demand for additional buffers through the input pad. A typical example of this situation is a parser that has not demanded enough bytes to parse the whole frame.

Native code integration

In Membrane we try to make use of Elixir goodness as often as possible. However, sometimes it is necessary to write some native code, for example to integrate with existing native libraries. To achieve that, we use natives, which are either NIFs or CNodes. Both of them have their drawbacks: CNodes introduce inter-process communication, while NIFs can even crash entire VM in case of failure. That's why we use them only when necessary and created some tools to make dealing with them easier.

Bundlex

To simplify and unify the process of writing and compiling natives, we use our own build tool - Bundlex. It is a multi-platform tool that provides a convenient way of compiling and accessing natives from Elixir code. For more information, see Bundlex's GitHub page.

Unifex

Process of creating natives is not only difficult but also quite arduous, because it requires using cumbersome Erlang APIs, and thus a lot of boilerplate code. To make it more pleasant, Membrane Framework provides Unifex, a tool that is responsible for generating interfaces between simple C libraries and Elixir on the base of short .exs configuration files. Unifex currently supports only NIFs, but support for CNodes is planned.

A quick introduction to Unifex is available here.

Membrane Common C

Membrane Framework also delivers some useful C routines which might be useful for creating natives. Membrane Common C package exports the following functionalities:

an interface to Membrane logger,

a ringbuffer,

implementation of Membrane.Payload.Behaviour wrapping Shmex) allowing to use shared memory segments as payload in Membrane elements.

Synchronization, clock and timers

Since version 0.4.0 Membrane Framework provides a means of synchronization between elements. This chapter presets how to use them: how to synchronize start of streams, create an element that is a clock provider and use a clock provided by a pipeline.

Synchronization

Membrane provides a sync mechanism to make sure 2 elements will start their streams at the same time. Good news - there's no need to modify the elements to support this feature! If the Membrane.ParentSpec contains info about elements to synchronize, the framework itself will ensure that they will receive StartOfStream events at the same moment.

The only change introduced by this mechanism is possibility to configure latency of an element. This is achieved via new action Membrane.Element.Action.latency_t/0 that can be returned from Membrane.Element.Base.handle_init/1 callback:

  @impl true
  def handle_init(options) do
    state = options |> Map.from_struct()
    {{:ok, latency: 20 |> Membrane.Time.milliseconds()}, state}
  end

This action is useful in a scenario where, for example, audio is sent over Bluetooth that introduces noticeable delay and video is presented on screen with almost no latency. In such scenario, the video player needs to be delayed to synchronize what you can see and hear. By setting latency you can let the framework to this for you!

Providing a clock

Some elements can provide a different source of time - like a hardware clock from sound card or elapsed time according to some library (like libshout consuming audio and sleeping for some time with small precision).

To become a clock provider for a pipeline your element needs to:

use Membrane.Element.Base.def_clock/1 to inform that this element exports clock

send updates to the clock process (Membrane.Clock.update_message_t/0) containing time to the next tick

Example

Let's consider a real-life example: PortAudio sink element. It plays audio to the devices in your system via PortAudio library - open-source wrapper for audio I/O APIs on different platforms. The element uses PortAudio API based on a callback - whenever PortAudio wants more data, it invokes the registered callback. So, to make sure audio samples produced by other elements in a pipeline, this element should export a clock in which the current time is based on the number of audio samples consumed by PortAudio.

First important thing is to notice is the def_clock macro invocation inside sink's module:

defmodule Membrane.Element.PortAudio.Sink do
  # ...

  def_clock """
  This clock measures time by counting a number of samples consumed by a PortAudio device
  and allows synchronization with the device.
  """

  # ...
end

With that macro present, during the initialization of the element, a clock process is spawned by the framework and its pid is available via :clock field in context for callbacks (other than handle_init).

This pid is then passed by the element to the native resource when the element enters :playing state

@impl true
  def handle_prepared_to_playing(ctx, state) do

    # ...

    with {:ok, {latency_ms, native}} <-
           SyncExecutor.apply(Native, :create, [
             self(),
             ctx.clock, # <-- A clock passed here
             endpoint_id,
             ringbuffer_size,
             pa_buffer_size,
             latency
           ]) do
    # ...

From this point, when PortAudio consumes audio it sends the update message to this clock.

Here's the snippet from the native part of PortAudio sink. It's a callback called by PortAudio when an output device demands more audio data. To reduce the number of messages sent, the actual message is sent every 100th invocation of the callback.

#define SAMPLES_PER_MSEC 48

static int callback(const void *_input_buffer, void *output_buffer,
                    unsigned long frames_per_buffer,
                    const PaStreamCallbackTimeInfo *_time_info,
                    PaStreamCallbackFlags _flags, void *user_data) {
  // ...

  SinkState *state = (SinkState *)user_data;

  if (++(state->ticks) % 100 == 0) {
    send_membrane_clock_update(env, state->clock, UNIFEX_SEND_THREADED,
                               100 * frames_per_buffer, SAMPLES_PER_MSEC);
  }

  // ...
}

The message format sent by send_membrane_clock_update is defined in sink.spec.exs. As you can see in the snippet below, the :membrane_clock_update message contains a tuple with a number of frames and number of samples per millisecond. We could divide frames by samples and get time in milliseconds, but instead we send both values (that can be interpreted as numerator and denominator) - this way the clock can ensure that division rounding error won't affect accuracy of the clock.

sends {:membrane_clock_update :: label, {frames :: int, sample_rate_ms :: int}}

The clock process accepts updates in different representations of time to next tick:

a single integer with time in milliseconds

tuple with numerator and denominator (used above)

rational number created by Ratio library (Ratio.t/0) - it can keep simplified fraction (2 integers) if needed to prevent rounding

They are described by Membrane.Clock.update_message_t/0 type.

Timers - using a clock

Each element can use the clock provided by a pipeline by setting up the timer Timer is a process that sends ticks in intervals set when it is started. This can be done by returning Membrane.Element.Action.start_timer_t/0 action that requires a tuple containing:

an atom - id for a new timer,

time interval between ticks in (that can be of type Membrane.Time.t/0 or Ratio.t/0 where the numerator is Membrane.Time.t/0)

(optionally) pid of a clock that should be used by the timer. If no clock provided, a pipeline clock is used.

Starting a clock means that the element will start receiving ticks that should be handled by a new callback - Membrane.Element.Base.handle_tick/3 receiving timer id along with context and state.

The timer can be stopped by Membrane.Element.Action.stop_timer_t/0 action with id of a timer.

Example

Here's a simplified example of a sink element that uses timer to consume video frames at the right speed.

It starts a timer on start_of_stream with id :demand_timer, interval being an inversion of framerate (that means if framerate is 30/1, the timer will send tick every 1/30th of a second, ~33 ms) and default (pipeline's) clock

Notice

Thanks to using Ratio.new/2 as an interval the rounding error will not accumulate. Since interval has to be an integer, we could provide it as 1 |> Time.second() |> div(30) but as the result would be rounded, the tick demanding 1 000 000 000th frame would be sent 333 ms earlier than it should:

alias Membrane.Time

iex> 1 |> Time.second() |> div(30) |> Kernel.*(1_000_000_000) |> Time.to_milliseconds()
33333333000

iex> 1 |> Time.second() |> Ratio.new(30) |> Ratio.*(1_000_000_000) \
|> Ratio.floor() |> Time.to_milliseconds()
33333333333

Demands a new frame on every tick (in handle_tick/3)

Stops the timer on either end_of_stream or when leaving :playing state

defmodule Membrane.Element.VideoSink do
  use Membrane.Sink

  alias Membrane.{Buffer, Time}
  alias Membrane.Caps.Video.Raw

  def_input_pad :input, caps: Raw, demand_unit: :buffers

  @impl true
  def handle_init(_) do
    {:ok, %{timer_started: false}}
  end

  @impl true
  def handle_playing_to_prepared(_ctx, %{timer_started: true} = state) do
    {{:ok, stop_timer: :timer}, %{state | timer_started: false}}
  end

  def handle_playing_to_prepared(_ctx, state) do
    {:ok, state}
  end

  @impl true
  def handle_start_of_stream(:input, ctx, state) do
    use Ratio
    {nom, denom} = ctx.pads.input.caps.framerate
    # framerate means we need `num` of frames each `denom` seconds
    demand_every = Ratio.new(Time.seconds(denom), nom)

    timer = {:demand_timer, demand_every}
    state = %{state | timer_started: true}

    {{:ok, demand: :input, start_timer: timer}, state}
  end

  @impl true
  def handle_end_of_stream(:input, _ctx, state) do
    {{:ok, stop_timer: :timer}, %{state | timer_started: false}}
  end

  @impl true
  def handle_write(:input, %Buffer{payload: _payload}, _ctx, state) do
    # ...
    # Code that does something with the payload, e.g. draw it on screen
    # ...
    {:ok, state}
  end

  @impl true
  def handle_tick(:demand_timer, _ctx, state) do
    {{:ok, demand: :input}, state}
  end
end